package mq

import (
	"bytes"
	"context"
	"fmt"
	"time"

	"go.uber.org/zap"

	// "bmember/pkg/mail"
	"gitee.com/quantum-force/of-utils/pkg/mail"

	"bmember/internal/model"

	"github.com/go-playground/validator/v10"
	jsoniter "github.com/json-iterator/go"
	"github.com/nsqio/go-nsq"

	"bmember/pkg/mq"
	"bmember/pkg/util"

	"bmember/internal/conf"
	"bmember/internal/dao"
	"bmember/pkg/log"
	"bmember/pkg/lueerror"
)

var json = jsoniter.ConfigCompatibleWithStandardLibrary

type MqBase struct {
	logger   log.Factory
	cfg      *conf.Config
	validate *validator.Validate
}

var (
	base *MqBase
)

// InitMqBase init mq base
func InitMqBase(logger log.Factory, cfg *conf.Config) {
	base = &MqBase{
		logger:   logger,
		cfg:      cfg,
		validate: validator.New(),
	}
}

// ParseBody 请求 body 解析
func (b *MqBase) ParseBody(msg []byte, model interface{}) lueerror.LueError {
	err := json.Unmarshal(msg, model)
	if err != nil {
		return lueerror.InvalidRequest(err)
	}
	if errV := b.validate.Struct(model); errV != nil {
		return lueerror.InvalidRequest(errV)
	}
	return nil
}

// 持久化失败消息
func (b *MqBase) saveFailedMsg(ctx context.Context, topic string, msg *nsq.Message) error {
	m := model.NewMessage(topic, msg.Body, model.MsgTypeForConsuming)
	_, err := dao.GetMessageDao().CreateMessage(ctx, m)
	return err
}

func (b *MqBase) stdLogger(ctx context.Context, topic string, start time.Time, msg *nsq.Message, err error) {
	body := msg.Body
	du := time.Since(start)
	if len(body) > 1024 {
		body = body[:1024]
	}
	if err != nil {
		b.logger.For(ctx).Error("failed",
			zap.String("topic", topic),
			zap.ByteString("message", body),
			zap.String("duration", du.String()),
			zap.String("error", err.Error()))
	} else {
		b.logger.For(ctx).Info("success",
			zap.String("topic", topic),
			zap.ByteString("message", body),
			zap.String("duration", du.String()))
	}
}

func (b *MqBase) sendMail(ctx context.Context, topic string, err error) {
	if err == nil {
		return
	}
	// 防止重复发送
	digest, _ := util.SignSha256(err.Error())
	if !dao.GetMessageDao().MailMsgCache(ctx, digest) {
		return
	}
	rID := ctx.Value("_requestID")
	requestID := ""
	if rID != nil {
		requestID, _ = rID.(string)
	}
	errStack := fmt.Errorf("  topic: %s\n\n  request_id: %s\n\n detail: %s\n\n ", topic, requestID, err.Error())
	mailMsg := &mail.Message{
		Subject:   "BMember MQ 异常",
		Content:   bytes.NewBufferString("<pre>" + errStack.Error() + "</pre>"),
		To:        []string{"backend@luedongtech.com"},
		Extension: nil,
	}
	_ = mail.AsyncSend(mailMsg, func(err error) {
		if err != nil {
			b.logger.For(ctx).Error(fmt.Sprintf("邮件发送错误: %v", err))
		}
	})
}

func (b *MqBase) wrap(topic string, handler mq.MqHandlerFunc) mq.MqHandlerFunc {
	return func(ctx context.Context, msg *nsq.Message) error {
		start := util.GetNow()
		digest, err := util.Sha256Hash(msg.Body)
		if err != nil {
			return err
		}
		if !dao.GetMessageDao().CanMsgBeConsumed(ctx, digest) {
			return nil
		}
		err = handler(ctx, msg)
		// 标准日志输出
		b.stdLogger(ctx, topic, start, msg, err)
		// 错误后续处理
		// 持久化消息
		if err != nil {
			// 错误邮件
			b.sendMail(ctx, topic, err)
			// 持久化失败消息
			err = b.saveFailedMsg(ctx, topic, msg)
		}
		// 持久化成功，也将消息标记为成功
		if err != nil {
			// 持久化失败，回滚消息幂等，返回错误，让 nsq 系统重发
			errT := dao.GetMessageDao().RollbackCacheConsumedMsg(ctx, digest)
			if errT != nil {
				b.logger.For(ctx).Error(fmt.Sprintf("cache msg error %s", errT.Error()))
			}
		}
		return err
	}
}
